package com.kgc.bigdata.hadoop.hdfs.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;

import java.net.URI;

/**
 * 作业：合并小文件
 */
public class MergeFile {

    public static void merge(String srcPath, String destPath) {
        try {
            // 读取hadoop文件系统的配置
            Configuration conf = new Configuration();
            String HDFSUri = "hdfs://hadoop000:8020";

            // 获取远端文件系统
            URI uri = new URI(HDFSUri);
            FileSystem remote = FileSystem.get(uri, conf);

            // 获得本地文件系统
            FileSystem local = FileSystem.getLocal(conf);

            // 获取data目录下的所有文件路径
            Path[] dirs = FileUtil.stat2Paths(local.globStatus(new Path(srcPath)));

            FSDataOutputStream out = null;
            FSDataInputStream in = null;

            for (Path dir : dirs) {
                // 文件名称
                String fileName = dir.getName().replace("-", "");
                // 只接受目录下的.txt文件
                FileStatus[] localStatus = local.globStatus(new Path(dir + "/*"), new RegexAcceptPathFilter("^.*.txt$"));
                // 获得目录下的所有文件
                Path[] listedPaths = FileUtil.stat2Paths(localStatus);
                // 输出路径
                Path block = new Path(destPath + "/" + fileName + ".txt");
                // 打开输出流
                out = remote.create(block);
                for (Path p : listedPaths) {
                    // 打开输入流
                    in = local.open(p);
                    // 复制数据
                    IOUtils.copyBytes(in, out, 4096, false);
                    // 关闭输入流
                    in.close();
                }
                if (out != null) {
                    // 关闭输出流
                    out.close();
                }
            }
        } catch (Exception e) {
            System.err.println(e);
        }
    }

    public static void main(String[] args) {
        String srcPath = "";
        String destPath = "";
        merge(srcPath, destPath);
    }
}

class RegexAcceptPathFilter implements PathFilter {
    private final String regex;

    public RegexAcceptPathFilter(String regex) {
        this.regex = regex;
    }

    @Override
    public boolean accept(Path path) {
        boolean flag = path.toString().matches(regex);
        return flag;
    }
}
